package com.bbx.flink.demo.windowfuncation;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import sun.awt.SunHints;

/**
 * 通过reduce 实现ProcessWindowFunction 的增量运算
 */
@Slf4j
public class ProcessWindowFunctionAndAggregation {

    public static void main(String [] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.socketTextStream("192.168.10.131",10003)
                .map((MapFunction<String ,Integer>) value->Integer.parseInt(value))
                .keyBy(i->i%2)
                .timeWindow(Time.minutes(1))
                //以下相当于 reduce(自定义的ReduceFunction，自定义的ProcessWindowFunction)，
                //个人理解，按照流式处理，先把数据在ReduceFunction 中进行增量聚合，然后把最终结果在ProcessWindowFunction 中获取
                .reduce(new ReduceFunction<Integer>() {
                    private static final long serialVersionUID = 437523617968801631L;

                    @Override
                    public Integer reduce(Integer value1, Integer value2) throws Exception {
                        log.info("~~~~~~~~~~~~~~~~~~~~~~~~~~~元素相加value1[{}]，value[{}]",value1,value2);
                        return value1+value2;
                    }
                }, new ProcessWindowFunction<Integer, Integer, Integer, TimeWindow>() {
                    private static final long serialVersionUID = 3679962978954407842L;

                    @Override
                    public void process(Integer integer, Context context, Iterable<Integer> elements, Collector<Integer> out) throws Exception {
                        //没太明白
                        Integer next = elements.iterator().next();
                        log.info("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~获取下一个元素[{}]",next);
                      out.collect(next);

                    }
                })
                .print();
        env.execute();
    }
}
